package com.atguigu.flink.window.all;

import com.atguigu.flink.pojo.WaterSensor;
import com.atguigu.flink.util.MyUtil;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.windowing.ProcessAllWindowFunction;
import org.apache.flink.streaming.api.windowing.windows.GlobalWindow;
import org.apache.flink.util.Collector;

/**
 * Created by Smexy on 2022/11/22
 *      countWindowXxx: 基于元素个数的窗口
 *
 *      countWindowAll(long size):  滚动窗口
 *      countWindowAll(long size, long slide)： 滑动窗口
 *
 *          没有会话窗口。
 *
 *      xxxAll:  要求窗口计算的并行度只能是1，keyBy和不KeyBy没区别！
 *                  加了keyBy无意义
 */
public class Demo1_CountWindow
{
    public static void main(String[] args) {


        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        env
           .socketTextStream("hadoop103", 8888)
           .map(new MapFunction<String, WaterSensor>()
           {
               @Override
               public WaterSensor map(String value) throws Exception {
                   String[] data = value.split(",");
                   return new WaterSensor(
                       data[0],
                       Long.valueOf(data[1]),
                       Integer.valueOf(data[2])
                   );
               }
           })
           //.keyBy(WaterSensor::getId)  //加了没意义
           //滚动窗口
           //.countWindowAll(3)
           //.countWindowAll(3,3)
           //滑动
           //.countWindowAll(3,5)  //滑的大，就会丢数据
           .countWindowAll(3,1)  //滑的小，就会出现1个数据落在多个窗口，重复算
           .process(new ProcessAllWindowFunction<WaterSensor, String, GlobalWindow>()
           {
               /*
                    Context context: 编程环境
                    Iterable<WaterSensor> elements： 进入窗口的所有元素的可迭代的集合
                    Collector<String> out： 输出什么
                */
               @Override
               public void process(Context context, Iterable<WaterSensor> elements, Collector<String> out) throws Exception {

                   out.collect(MyUtil.parseList(elements).toString());
               }
           }).setParallelism(10)
           .print();

        try {
                    env.execute();
                } catch (Exception e) {
                    e.printStackTrace();
                }

    }
}
